package com.zhang.third.day03;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 * @title:
 * @author: zhang
 * @date: 2022/4/6 12:08
 */
public class Example9 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .socketTextStream("localhost", 9999)
                .keyBy(r -> 1)
                .process(new KeyedProcessFunction<Integer, String, String>() {
                    @Override
                    public void processElement(String value, KeyedProcessFunction<Integer, String, String>.Context ctx, Collector<String> out) throws Exception {
                        // 获取数据触发processElement函数的系统时间
                        long currTs = ctx.timerService().currentProcessingTime();
                        // 30s之后的时间戳
                        long thirtySecondsLater = currTs + 30 * 1000L;
                        // 注册一个30s之后触发的定时器
                        ctx.timerService().registerProcessingTimeTimer(thirtySecondsLater);
                        out.collect("数据：" + value + "到达时间是：" + new Timestamp(currTs) + "," +
                                "注册了一个时间戳是：" + new Timestamp(thirtySecondsLater) + "的定时器！");
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<Integer, String, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        // timestamp参数是注册定时器的时间戳，也就是thirtySecondsLater
                        out.collect("定时器时间戳是："+new Timestamp(timestamp)+",触发了！" +
                                "定时器真正执行的系统时间是："+new Timestamp(ctx.timerService().currentProcessingTime()));
                    }
                })
                .print();

        env.execute();
    }
}
